//// THIS CODE AND INFORMATION IS PROVIDED "AS IS" WITHOUT WARRANTY OF
//// ANY KIND, EITHER EXPRESSED OR IMPLIED, INCLUDING BUT NOT LIMITED TO
//// THE IMPLIED WARRANTIES OF MERCHANTABILITY AND/OR FITNESS FOR A
//// PARTICULAR PURPOSE.
////
//// Copyright (c) Microsoft Corporation. All rights reserved

#include "pch.h"
#include "FStreamSink.h"
#include "FMediaSink.h"
#include "Helpers.h"
#include "VideoSampleProcessor.h"
#include "AudioSampleProcessor.h"
#include "FragmentBuilder.h"

using namespace Microsoft::Media::FragmentedMediaSink;

CStreamSink::CStreamSink(DWORD dwIdentifier)
	: _cRef(1)
	, _dwIdentifier(dwIdentifier)
	, _state(State_TypeNotSet)
	, _IsShutdown(false)
	, _fIsVideo(false)
	, _WorkQueueId(0)
	, _pParent(nullptr)
#pragma warning(push)
#pragma warning(disable:4355)
	, _WorkQueueCB(this, &CStreamSink::OnDispatchWorkItem)
#pragma warning(pop)
{
	ZeroMemory(&_guiCurrentSubtype, sizeof(_guiCurrentSubtype));
}

CStreamSink::~CStreamSink()
{
	assert(_IsShutdown);
}


// IUnknown methods

IFACEMETHODIMP CStreamSink::QueryInterface(REFIID riid, void **ppv)
{
	if (ppv == nullptr)
	{
		return E_POINTER;
	}
	(*ppv) = nullptr;

	HRESULT hr = S_OK;
	if (riid == IID_IUnknown ||
		riid == IID_IMFStreamSink ||
		riid == IID_IMFMediaEventGenerator)
	{
		(*ppv) = static_cast<IMFStreamSink*>(this);
		AddRef();
	}
	else if (riid == IID_IMFMediaTypeHandler)
	{
		(*ppv) = static_cast<IMFMediaTypeHandler*>(this);
		AddRef();
	}
	else
	{
		hr = E_NOINTERFACE;
	}

	if (FAILED(hr) && riid == IID_IMarshal)
	{
		if (_spFTM == nullptr)
		{
			AutoLock lock(_critSec);
			if (_spFTM == nullptr)
			{
				hr = CoCreateFreeThreadedMarshaler(static_cast<IMFStreamSink*>(this), &_spFTM);
			}
		}

		if (SUCCEEDED(hr))
		{
			if (_spFTM == nullptr)
			{
				hr = E_UNEXPECTED;
			}
			else
			{
				hr = _spFTM.Get()->QueryInterface(riid, ppv);
			}
		}
	}

	return hr;
}

IFACEMETHODIMP_(ULONG) CStreamSink::AddRef()
{
	return InterlockedIncrement(&_cRef);
}

IFACEMETHODIMP_(ULONG) CStreamSink::Release()
{
	long cRef = InterlockedDecrement(&_cRef);
	if (cRef == 0)
	{
		delete this;
	}
	return cRef;
}

// IMFMediaEventGenerator methods.
// Note: These methods call through to the event queue helper object.

IFACEMETHODIMP CStreamSink::BeginGetEvent(IMFAsyncCallback *pCallback, IUnknown *punkState)
{
	HRESULT hr = S_OK;

	AutoLock lock(_critSec);

	hr = CheckShutdown();

	if (SUCCEEDED(hr))
	{
		hr = _spEventQueue->BeginGetEvent(pCallback, punkState);
	}

	TRACEHR_RET(hr);
}

IFACEMETHODIMP CStreamSink::EndGetEvent(IMFAsyncResult *pResult, IMFMediaEvent **ppEvent)
{
	HRESULT hr = S_OK;

	AutoLock lock(_critSec);

	hr = CheckShutdown();

	if (SUCCEEDED(hr))
	{
		hr = _spEventQueue->EndGetEvent(pResult, ppEvent);
	}

	TRACEHR_RET(hr);
}

IFACEMETHODIMP CStreamSink::GetEvent(DWORD dwFlags, IMFMediaEvent **ppEvent)
{
	// NOTE:
	// GetEvent can block indefinitely, so we don't hold the lock.
	// This requires some juggling with the event queue pointer.

	HRESULT hr = S_OK;

	ComPtr<IMFMediaEventQueue> spQueue;

	{
		AutoLock lock(_critSec);

		// Check shutdown
		hr = CheckShutdown();

		// Get the pointer to the event queue.
		if (SUCCEEDED(hr))
		{
			spQueue = _spEventQueue;
		}
	}

	// Now get the event.
	if (SUCCEEDED(hr))
	{
		hr = spQueue->GetEvent(dwFlags, ppEvent);
	}

	TRACEHR_RET(hr);
}

IFACEMETHODIMP CStreamSink::QueueEvent(MediaEventType met, REFGUID guidExtendedType, HRESULT hrStatus, PROPVARIANT const *pvValue)
{
	HRESULT hr = S_OK;

	AutoLock lock(_critSec);

	hr = CheckShutdown();

	if (SUCCEEDED(hr))
	{
		hr = _spEventQueue->QueueEventParamVar(met, guidExtendedType, hrStatus, pvValue);
	}

	TRACEHR_RET(hr);
}

/// IMFStreamSink methods

IFACEMETHODIMP CStreamSink::GetMediaSink(IMFMediaSink **ppMediaSink)
{
	if (ppMediaSink == nullptr)
	{
		return E_INVALIDARG;
	}

	AutoLock lock(_critSec);

	HRESULT hr = CheckShutdown();

	if (SUCCEEDED(hr))
	{
		_spSink.Get()->QueryInterface(IID_IMFMediaSink, (void**)ppMediaSink);
	}

	TRACEHR_RET(hr);
}

IFACEMETHODIMP CStreamSink::GetIdentifier(DWORD *pdwIdentifier)
{
	if (pdwIdentifier == nullptr)
	{
		return E_INVALIDARG;
	}

	AutoLock lock(_critSec);

	HRESULT hr = CheckShutdown();

	if (SUCCEEDED(hr))
	{
		*pdwIdentifier = _dwIdentifier;
	}

	TRACEHR_RET(hr);
}

IFACEMETHODIMP CStreamSink::GetMediaTypeHandler(IMFMediaTypeHandler **ppHandler)
{
	if (ppHandler == nullptr)
	{
		return E_INVALIDARG;
	}

	AutoLock lock(_critSec);

	HRESULT hr = CheckShutdown();

	// This stream object acts as its own type handler, so we QI ourselves.
	if (SUCCEEDED(hr))
	{
		hr = QueryInterface(IID_IMFMediaTypeHandler, (void**)ppHandler);
	}

	TRACEHR_RET(hr);
}

// We received a sample from an upstream component
IFACEMETHODIMP CStreamSink::ProcessSample(IMFSample *pSample)
{
	if (pSample == nullptr)
	{
		return E_INVALIDARG;
	}

	HRESULT hr = S_OK;

	AutoLock lock(_critSec);

	hr = CheckShutdown();

	// Validate the operation.
	if (SUCCEEDED(hr))
	{
		hr = ValidateOperation(OpProcessSample);
	}

	// Add the sample to the sample queue.
	if (SUCCEEDED(hr))
	{
		hr = _SampleQueue.InsertBack(pSample);
	}

	// Unless we are paused, start an async operation to dispatch the next sample.
	if (SUCCEEDED(hr))
	{
		if (_state != State_Paused)
		{
			// Queue the operation.
			hr = QueueAsyncOperation(OpProcessSample);
		}
	}

	TRACEHR_RET(hr);
}

// The client can call PlaceMarker at any time. In response,
// we need to queue an MEStreamSinkMarker event, but not until
// *after *we have processed all samples that we have received
// up to this point.
//
// Also, in general you might need to handle specific marker
// types, although this sink does not.

IFACEMETHODIMP CStreamSink::PlaceMarker(
	MFSTREAMSINK_MARKER_TYPE eMarkerType,
	const PROPVARIANT *pvarMarkerValue,
	const PROPVARIANT *pvarContextValue)
{
	AutoLock lock(_critSec);

	HRESULT hr = S_OK;
	ComPtr<IMarker> spMarker;

	hr = CheckShutdown();

	if (SUCCEEDED(hr))
	{
		hr = ValidateOperation(OpPlaceMarker);
	}

	if (SUCCEEDED(hr))
	{
		hr = CreateMarker(eMarkerType, pvarMarkerValue, pvarContextValue, &spMarker);
	}

	if (SUCCEEDED(hr))
	{
		hr = _SampleQueue.InsertBack(spMarker.Get());
	}

	// Unless we are paused, start an async operation to dispatch the next sample/marker.
	if (SUCCEEDED(hr))
	{
		if (_state != State_Paused)
		{
			// Queue the operation.
			hr = QueueAsyncOperation(OpPlaceMarker); // Increments ref count on pOp.
		}
	}

	TRACEHR_RET(hr);
}

// Discards all samples that were not processed yet.
IFACEMETHODIMP CStreamSink::Flush()
{
	AutoLock lock(_critSec);
	HRESULT hr = S_OK;
	try
	{
		ThrowIfError(CheckShutdown());

		// Note: Even though we are flushing data, we still need to send
		// any marker events that were queued.
		DropSamplesFromQueue();
	}
	catch (Exception ^exc)
	{
		hr = exc->HResult;
	}

	TRACEHR_RET(hr);
}


/// IMFMediaTypeHandler methods

// Check if a media type is supported.
IFACEMETHODIMP CStreamSink::IsMediaTypeSupported(
	/* [in] */ IMFMediaType *pMediaType,
	/* [out] */ IMFMediaType **ppMediaType)
{
	if (pMediaType == nullptr)
	{
		return E_INVALIDARG;
	}

	AutoLock lock(_critSec);

	GUID majorType = GUID_NULL;

	HRESULT hr = CheckShutdown();

	if (SUCCEEDED(hr))
	{
		hr = pMediaType->GetGUID(MF_MT_MAJOR_TYPE, &majorType);
	}

	// First make sure it's video or audio type.
	if (SUCCEEDED(hr))
	{
		if (majorType != MFMediaType_Video && majorType != MFMediaType_Audio)
		{
			hr = MF_E_INVALIDTYPE;
		}
	}

	if (SUCCEEDED(hr) && _spCurrentType != nullptr)
	{
		GUID guiNewSubtype;
		if (FAILED(pMediaType->GetGUID(MF_MT_SUBTYPE, &guiNewSubtype)) ||
			guiNewSubtype != _guiCurrentSubtype)
		{
			hr = MF_E_INVALIDTYPE;
		}
	}

	// We don't return any "close match" types.
	if (ppMediaType)
	{
		*ppMediaType = nullptr;
	}

	TRACEHR_RET(hr);
}


// Return the number of preferred media types.
IFACEMETHODIMP CStreamSink::GetMediaTypeCount(DWORD *pdwTypeCount)
{
	if (pdwTypeCount == nullptr)
	{
		return E_INVALIDARG;
	}

	AutoLock lock(_critSec);

	HRESULT hr = CheckShutdown();

	if (SUCCEEDED(hr))
	{
		// We've got only one media type
		*pdwTypeCount = 1;
	}

	TRACEHR_RET(hr);
}


// Return a preferred media type by index.
IFACEMETHODIMP CStreamSink::GetMediaTypeByIndex(
	/* [in] */ DWORD dwIndex,
	/* [out] */ IMFMediaType **ppType)
{
	if (ppType == nullptr)
	{
		return E_INVALIDARG;
	}

	AutoLock lock(_critSec);

	HRESULT hr = CheckShutdown();

	if (dwIndex > 0)
	{
		hr = MF_E_NO_MORE_TYPES;
	}
	else
	{
		*ppType = _spCurrentType.Get();
		if (*ppType != nullptr)
		{
			(*ppType)->AddRef();
		}
	}

	TRACEHR_RET(hr);
}


// Set the current media type.
IFACEMETHODIMP CStreamSink::SetCurrentMediaType(IMFMediaType *pMediaType)
{
	HRESULT hr = S_OK;
	try
	{
		if (pMediaType == nullptr)
		{
			Throw(E_INVALIDARG);
		}
		AutoLock lock(_critSec);

		ThrowIfError(CheckShutdown());

		// We don't allow format changes after streaming starts.
		ThrowIfError(ValidateOperation(OpSetMediaType));

		// We set media type already
		if (_state >= State_Ready)
		{
			ThrowIfError(IsMediaTypeSupported(pMediaType, nullptr));
		}

		GUID guiMajorType;
		pMediaType->GetMajorType(&guiMajorType);
		_fIsVideo = (guiMajorType == MFMediaType_Video);

		ThrowIfError(MFCreateMediaType(_spCurrentType.ReleaseAndGetAddressOf()));
		ThrowIfError(pMediaType->CopyAllItems(_spCurrentType.Get()));
		ThrowIfError(_spCurrentType->GetGUID(MF_MT_SUBTYPE, &_guiCurrentSubtype));

		if (_state < State_Ready)
		{
			_state = State_Ready;
		}
		else if (_state > State_Ready)
		{
			ComPtr<IMFMediaType> spType;
			ThrowIfError(MFCreateMediaType(&spType));
			ThrowIfError(pMediaType->CopyAllItems(spType.Get()));
			ProcessFormatChange(spType.Get());
		}
	}
	catch (Exception ^exc)
	{
		hr = exc->HResult;
	}
	TRACEHR_RET(hr);
}

// Return the current media type, if any.
IFACEMETHODIMP CStreamSink::GetCurrentMediaType(IMFMediaType **ppMediaType)
{
	if (ppMediaType == nullptr)
	{
		return E_INVALIDARG;
	}

	AutoLock lock(_critSec);

	HRESULT hr = CheckShutdown();

	if (SUCCEEDED(hr))
	{
		if (_spCurrentType == nullptr)
		{
			hr = MF_E_NOT_INITIALIZED;
		}
	}

	if (SUCCEEDED(hr))
	{
		*ppMediaType = _spCurrentType.Get();
		(*ppMediaType)->AddRef();
	}

	TRACEHR_RET(hr);
}


// Return the major type GUID.
IFACEMETHODIMP CStreamSink::GetMajorType(GUID *pguidMajorType)
{
	if (pguidMajorType == nullptr)
	{
		return E_INVALIDARG;
	}

	if (!_spCurrentType)
	{
		return MF_E_NOT_INITIALIZED;
	}

	*pguidMajorType = (_fIsVideo) ? MFMediaType_Video : MFMediaType_Audio;

	return S_OK;
}

// private methods
HRESULT CStreamSink::Initialize(CMediaSink *pParent)
{
	assert(pParent != nullptr);

	HRESULT hr = S_OK;

	// Create the event queue helper.
	hr = MFCreateEventQueue(&_spEventQueue);

	// Allocate a new work queue for async operations.
	if (SUCCEEDED(hr))
	{
		hr = MFAllocateSerialWorkQueue(MFASYNC_CALLBACK_QUEUE_STANDARD, &_WorkQueueId);
	}

	if (SUCCEEDED(hr))
	{
		_spSink = pParent;
		_pParent = pParent;
	}

	TRACEHR_RET(hr);
}

HRESULT CStreamSink::SetFragmentControllers(Windows::Foundation::Collections::IVector<Microsoft::Media::FragmentedMediaSink::IFragmentController^>^ fragmentControllers)
{
	assert(fragmentControllers != nullptr);

	HRESULT hr = S_OK;

	if (SUCCEEDED(hr))
	{		
		if (_fIsVideo)
		{
			_sampleProcessor = std::make_unique<VideoSampleProcessor>(fragmentControllers, _spCurrentType.Get(), _dwIdentifier);
		}
		else
		{
			_sampleProcessor = std::make_unique<AudioSampleProcessor>(fragmentControllers, _spCurrentType.Get(), _dwIdentifier);
		}
	}

	TRACEHR_RET(hr);
}

// Called when the presentation clock starts.
HRESULT CStreamSink::Start(MFTIME start)
{
	AutoLock lock(_critSec);

	HRESULT hr = S_OK;

	hr = ValidateOperation(OpStart);

	if (SUCCEEDED(hr))
	{
		_state = State_Started;
		hr = QueueAsyncOperation(OpStart);
	}

	TRACEHR_RET(hr);
}

// Called when the presentation clock stops.
HRESULT CStreamSink::Stop()
{
	AutoLock lock(_critSec);

	HRESULT hr = S_OK;

	hr = ValidateOperation(OpStop);

	if (SUCCEEDED(hr))
	{
		_state = State_Stopped;
		hr = QueueAsyncOperation(OpStop);
	}

	TRACEHR_RET(hr);
}

// Called when the presentation clock pauses.
HRESULT CStreamSink::Pause()
{
	AutoLock lock(_critSec);

	HRESULT hr = S_OK;

	hr = ValidateOperation(OpPause);

	if (SUCCEEDED(hr))
	{
		_state = State_Paused;
		hr = QueueAsyncOperation(OpPause);
	}

	TRACEHR_RET(hr);
}

// Called when the presentation clock restarts.
HRESULT CStreamSink::Restart()
{
	AutoLock lock(_critSec);

	HRESULT hr = S_OK;

	hr = ValidateOperation(OpRestart);

	if (SUCCEEDED(hr))
	{
		_state = State_Started;
		hr = QueueAsyncOperation(OpRestart);
	}

	TRACEHR_RET(hr);
}

// Class-static matrix of operations vs states.
// If an entry is TRUE, the operation is valid from that state.
BOOL CStreamSink::ValidStateMatrix[CStreamSink::State_Count][CStreamSink::Op_Count] =
{
	// States:    Operations:
	//            SetType   Start     Restart   Pause     Stop      Sample    Marker   
	/* NotSet */  TRUE, FALSE, FALSE, FALSE, FALSE, FALSE, FALSE,

	/* Ready */   TRUE, TRUE, FALSE, TRUE, TRUE, FALSE, TRUE,

	/* Start */   TRUE, TRUE, FALSE, TRUE, TRUE, TRUE, TRUE,

	/* Pause */   TRUE, TRUE, TRUE, TRUE, TRUE, TRUE, TRUE,

	/* Stop */    TRUE, TRUE, FALSE, FALSE, TRUE, FALSE, TRUE,

};

// Checks if an operation is valid in the current state.
HRESULT CStreamSink::ValidateOperation(StreamOperation op)
{
	assert(!_IsShutdown);

	if (ValidStateMatrix[_state][op])
	{
		return S_OK;
	}
	else if (_state == State_TypeNotSet)
	{
		return MF_E_NOT_INITIALIZED;
	}
	else
	{
		return MF_E_INVALIDREQUEST;
	}
}

// Shuts down the stream sink.
HRESULT CStreamSink::Shutdown()
{
	AutoLock lock(_critSec);

	if (!_IsShutdown)
	{
		if (_spEventQueue)
		{
			_spEventQueue->Shutdown();
		}

		MFUnlockWorkQueue(_WorkQueueId);

		_SampleQueue.Clear();

		_spSink.Reset();
		_spEventQueue.Reset();
		_spCurrentType.Reset();

		_IsShutdown = true;
	}

	return S_OK;
}


// Puts an async operation on the work queue.
HRESULT CStreamSink::QueueAsyncOperation(StreamOperation op)
{
	HRESULT hr = S_OK;
	ComPtr<CAsyncOperation> spOp;
	spOp.Attach(new CAsyncOperation(op)); // Created with ref count = 1
	if (!spOp)
	{
		hr = E_OUTOFMEMORY;
	}

	if (SUCCEEDED(hr))
	{
		hr = MFPutWorkItem2(_WorkQueueId, 0, &_WorkQueueCB, spOp.Get());
	}

	TRACEHR_RET(hr);
}

HRESULT CStreamSink::OnDispatchWorkItem(IMFAsyncResult *pAsyncResult)
{
	// Called by work queue thread. Need to hold the critical section.
	AutoLock lock(_critSec);

	try
	{
		ComPtr<IUnknown> spState;

		ThrowIfError(pAsyncResult->GetState(&spState));

		// The state object is a CAsncOperation object.
		CAsyncOperation *pOp = static_cast<CAsyncOperation *>(spState.Get());
		StreamOperation op = pOp->m_op;

		switch (op)
		{
		case OpStart:
		case OpRestart:
			// Send MEStreamSinkStarted.
			ThrowIfError(QueueEvent(MEStreamSinkStarted, GUID_NULL, S_OK, nullptr));

			// There might be samples queue from earlier (ie, while paused).
			bool fRequestMoreSamples;
			fRequestMoreSamples = SendSampleFromQueue();

			if (fRequestMoreSamples)
			{
				// If false there is no samples in the queue now so request one
				ThrowIfError(QueueEvent(MEStreamSinkRequestSample, GUID_NULL, S_OK, nullptr));
			}
			break;

		case OpStop:
			// Flush samples from queue.
			DropSamplesFromQueue();

			// Send the event even if the previous call failed.
			ThrowIfError(QueueEvent(MEStreamSinkStopped, GUID_NULL, S_OK, nullptr));
			break;

		case OpPause:
			ThrowIfError(QueueEvent(MEStreamSinkPaused, GUID_NULL, S_OK, nullptr));
			break;

		case OpProcessSample:
		case OpPlaceMarker:
		case OpSetMediaType:
			DispatchProcessSample(pOp);
			break;

		}
	}
	catch (Exception ^exc)
	{
		HandleError(exc->HResult);
	}
	return S_OK;
}

// Complete a ProcessSample or PlaceMarker request.
void CStreamSink::DispatchProcessSample(CAsyncOperation *pOp)
{
	assert(pOp != nullptr);
	bool fRequestMoreSamples = false;
	fRequestMoreSamples = SendSampleFromQueue();

	// Ask for another sample
	if (fRequestMoreSamples)
	{
		if (pOp->m_op == OpProcessSample)
		{
			ThrowIfError(QueueEvent(MEStreamSinkRequestSample, GUID_NULL, S_OK, nullptr));
		}
	}
}

// Drop samples in the queue
bool CStreamSink::DropSamplesFromQueue()
{
	_sampleProcessor->Flush();
	ProcessSamplesFromQueue(true);

	return true;
}

// Send sample from the queue
bool CStreamSink::SendSampleFromQueue()
{
	return ProcessSamplesFromQueue(false);
}

bool CStreamSink::ProcessSamplesFromQueue(bool fFlush)
{
	bool fNeedMoreSamples = false;

	ComPtr<IUnknown> spunkSample;

	bool fSendSamples = true;
	bool fSendEOS = false;

	if (FAILED(_SampleQueue.RemoveFront(&spunkSample)))
	{
		fNeedMoreSamples = true;
		fSendSamples = false;
	}

	while (fSendSamples)
	{
		ComPtr<IMFSample> spSample;
		//ComPtr<IMFSample> spPacket;
		bool fProcessingSample = false;
		assert(spunkSample);

		// Figure out if this is a marker or a sample.
		// If this is a sample, write it to the sink writer (FragmentBuilder).
		// Now handle the sample/marker appropriately.
		if (SUCCEEDED(spunkSample.As(&spSample)))
		{
			assert(spSample);    // Not a marker, must be a sample

			if (!fFlush)
			{
				// Prepare sample for sending
				//spPacket = PrepareSample(spSample.Get());
				fProcessingSample = true;
			}
		}
		else
		{
			ComPtr<IMarker> spMarker;
			// Check if it is a marker
			if (SUCCEEDED(spunkSample.As(&spMarker)))
			{
				MFSTREAMSINK_MARKER_TYPE markerType;
				PROPVARIANT var;
				PropVariantInit(&var);
				ThrowIfError(spMarker->GetMarkerType(&markerType));
				// Get the context data.
				ThrowIfError(spMarker->GetContext(&var));

				HRESULT hr = QueueEvent(MEStreamSinkMarker, GUID_NULL, S_OK, &var);

				PropVariantClear(&var);

				ThrowIfError(hr);

				if (markerType == MFSTREAMSINK_MARKER_ENDOFSEGMENT)
				{
					fSendEOS = true;
				}
				else if (markerType == MFSTREAMSINK_MARKER_TICK)
				{
					if (!fFlush)
					{
						// send a tick to indicate a blank spot in the stream
						// The marker data is a LONGLONG value (VT_I8) that specifies the time for the missing sample.
						PROPVARIANT varValue;
						PropVariantInit(&varValue);
						spMarker->GetMarkerValue(&varValue);
						if (varValue.vt == VT_I8)
						{
							_sampleProcessor->SendTick(varValue.hVal.QuadPart);
						}
						PropVariantClear(&varValue);
					}
				}
			}
			else
			{
				ComPtr<IMFMediaType> spType;
				ThrowIfError(spunkSample.As(&spType));
				if (!fFlush)
				{
					//spPacket = PrepareFormatChange(spType.Get());
				}
			}
		}

		if (spSample)
		{
			ThrowIfError(_sampleProcessor->ProcessSample(spSample.Get()));

			if (_state == State_Started && fProcessingSample)
			{
				// If we are still in started state request another sample
				ThrowIfError(QueueEvent(MEStreamSinkRequestSample, GUID_NULL, S_OK, nullptr));
			}
			// We stop if we processed a sample otherwise keep looking
			fSendSamples = !fProcessingSample;
		}

		if (fSendSamples)
		{
			if (FAILED(_SampleQueue.RemoveFront(spunkSample.ReleaseAndGetAddressOf())))
			{
				fNeedMoreSamples = true;
				fSendSamples = false;
			}
		}

	}

	if (fSendEOS)
	{
		HRESULT hr = _sampleProcessor->Finalize();
		if (FAILED(hr))
		{
			_pParent->ReportNonCriticalError(ref new Exception(hr));
		}
		ComPtr<CMediaSink> spParent = _pParent;
		concurrency::create_task([spParent, this](){
			spParent->ReportEndOfStream();
		});
	}

	return fNeedMoreSamples;
}

// Processing format change
void CStreamSink::ProcessFormatChange(IMFMediaType *pMediaType)
{
	assert(pMediaType != nullptr);

	// Add the media type to the sample queue.
	ThrowIfError(_SampleQueue.InsertBack(pMediaType));

	// Unless we are paused, start an async operation to dispatch the next sample.
	// Queue the operation.
	ThrowIfError(QueueAsyncOperation(OpSetMediaType));
}

CStreamSink::CAsyncOperation::CAsyncOperation(StreamOperation op)
	: _cRef(1)
	, m_op(op)
{
}

CStreamSink::CAsyncOperation::~CAsyncOperation()
{
	assert(_cRef == 0);
}

ULONG CStreamSink::CAsyncOperation::AddRef()
{
	return InterlockedIncrement(&_cRef);
}

ULONG CStreamSink::CAsyncOperation::Release()
{
	ULONG cRef = InterlockedDecrement(&_cRef);
	if (cRef == 0)
	{
		delete this;
	}

	return cRef;
}

HRESULT CStreamSink::CAsyncOperation::QueryInterface(REFIID iid, void **ppv)
{
	if (!ppv)
	{
		return E_POINTER;
	}
	if (iid == IID_IUnknown)
	{
		*ppv = static_cast<IUnknown*>(this);
	}
	else
	{
		*ppv = nullptr;
		return E_NOINTERFACE;
	}
	AddRef();
	return S_OK;
}

void CStreamSink::HandleError(HRESULT hr)
{
	if (!_IsShutdown)
	{
		QueueEvent(MEError, GUID_NULL, hr, nullptr);
	}
}